package main

import (
	"../common"
	"fmt"
	"regexp"
	"strings"
	"sync"
	"time"
)

type SenderStatics struct {
	MessageCount  int64 `json:"message_count"`
	FromTimestamp int64 `json:"from"`
	ToTimestamp   int64 `json:"to"`
	ErrorCount    int64 `json:"error_count"`
}

func (c *SenderStatics) MessageInc() {
	c.MessageCount++
}

func (c *SenderStatics) ErrorInc() {
	c.ErrorCount++
}

func (c *SenderStatics) Done() {
	c.ToTimestamp = time.Now().Unix()
}

type ConnStatics struct {
	Address        string `json:"-"`
	LastConnTime   int64  `json:"last_connect_time"`
	LastDisconTime int64  `json:"last_disconnect_time"`
	ConnCount      int64  `json:"connect_count"`
	DisconnCount   int64  `json:"disconnect_count"`
}

func (c *ConnStatics) Add(other *ConnStatics) {
	c.LastConnTime = bigger(c.LastConnTime, other.LastConnTime)
	c.LastDisconTime = bigger(c.LastDisconTime, other.LastDisconTime)
	c.ConnCount += other.ConnCount
	c.DisconnCount += other.DisconnCount
}

func bigger(a, b int64) int64 {
	if a < b {
		return b
	}
	return a
}

type MessageStatics struct {
	Address            string `json:"-"`
	ContactTime        int64  `json:"last_contact_time"`
	MessageTime        int64  `json:"last_message_time"`
	NormalMessageCount int64  `json:"normal_message_count"`
	ErrorMessageCount  int64  `json:"error_message_count"`
}

func (m *MessageStatics) String() string {
	return fmt.Sprintf("ContactTime:%d,MessageTime:%d,NormalMessageCount:%d,ErrorMessageCount:%d",
		m.ContactTime, m.MessageTime, m.NormalMessageCount, m.ErrorMessageCount)
}

func (m *MessageStatics) Add(other *MessageStatics) {
	m.ContactTime = bigger(m.ContactTime, other.ContactTime)
	m.MessageTime = bigger(m.MessageTime, other.MessageTime)
	m.NormalMessageCount += other.NormalMessageCount
	m.ErrorMessageCount += other.ErrorMessageCount
}

type Counter struct {
	staticsCh     chan *MessageStatics
	connCh        chan *ConnStatics
	mu            sync.RWMutex
	senderSummary []SenderStatics
	statics       map[string]*MessageStatics
	conns         map[string]*ConnStatics
	stopping      chan interface{}
	stopped       chan interface{}
}

func (c *Counter) QueryConnsByIP(addr string) map[string]ConnStatics {
	ret := make(map[string]ConnStatics)
	c.mu.RLock()
	defer c.mu.RUnlock()
	for k, v := range c.conns {
		if match, err := regexp.MatchString(addr, k); match && err == nil {
			ret[k] = *v
		}
	}
	return ret
}

func (c *Counter) QuerySenderSummary() []SenderStatics {
	return c.senderSummary
}

func (c *Counter) QueryStaticsByIP(addr string) map[string]MessageStatics {
	ret := make(map[string]MessageStatics)
	c.mu.RLock()
	defer c.mu.RUnlock()
	for k, v := range c.statics {
		if match, err := regexp.MatchString(addr, k); match && err == nil {
			ret[k] = *v
		}
	}
	return ret
}

func (c *Counter) OnSender(info *SenderStatics) {
	info.Done()
	c.mu.Lock()
	defer c.mu.Unlock()
	if len(c.senderSummary) < 10 {
		c.senderSummary = append(c.senderSummary, *info)
	} else {
		c.senderSummary = append(c.senderSummary[1:], *info)
	}
}

func (c *Counter) OnConnect(addr string) {
	c.connCh <- &ConnStatics{Address: addr, LastConnTime: time.Now().Unix(), ConnCount: 1}
}

func (c *Counter) OnDisconnect(addr string) {
	c.connCh <- &ConnStatics{Address: addr, LastDisconTime: time.Now().Unix(), DisconnCount: 1}
}

func (c *Counter) MessageInc(addr string, name string, ts int64) {
	actualAddr,err := common.ExtractMetricAddr(addr)
	if err!=nil{
        return
    }
	if actualAddr == "" {
		actualAddr = addr
	}
	msg := MessageStatics{Address: actualAddr, ContactTime: time.Now().Unix()}
	if ts != 0 && name != "" {
		msg.MessageTime = int64(ts)
		msg.NormalMessageCount = 1
	} else {
		msg.ErrorMessageCount = 1
	}
	c.staticsCh <- &msg
}

func (c *Counter) addMessage(msg *MessageStatics) {
	c.mu.Lock()
	defer c.mu.Unlock()
	addr := msg.Address
	s := strings.Split(addr, ":")
	if len(s) > 0 {
		addr = s[0]
	}
	if _, ok := c.statics[addr]; !ok {
		c.statics[addr] = &MessageStatics{}
	}
	c.statics[addr].Add(msg)
}

func (c *Counter) addConn(conn *ConnStatics) {
	c.mu.Lock()
	defer c.mu.Unlock()
	addr := conn.Address
	s := strings.Split(addr, ":")
	if len(s) > 0 {
		addr = s[0]
	}
	if _, ok := c.conns[addr]; !ok {
		c.conns[addr] = &ConnStatics{}
	}
	c.conns[addr].Add(conn)
}

func (c *Counter) Run() {
	for {
		select {
		case st, ok := <-c.staticsCh:
			if ok {
				c.addMessage(st)
			}
		case conn, ok := <-c.connCh:
			if ok {
				c.addConn(conn)
			}
		case <-c.stopping:
			close(c.stopped)
			return
		}
	}
}

func (c *Counter) Stop() {
	close(c.stopping)
	<-c.stopped
}

func NewCounter() *Counter {
	c := Counter{
		staticsCh: make(chan *MessageStatics),
		connCh:    make(chan *ConnStatics),
		statics:   make(map[string]*MessageStatics),
		conns:     make(map[string]*ConnStatics),
		stopping:  make(chan interface{}),
		stopped:   make(chan interface{}),
	}
	return &c
}
